package rabbitmq_demo

import (
	"context"
	"sync"

	"gitee.com/lihaiping1603/utils/log"
	amqp "github.com/rabbitmq/amqp091-go"
)

type DeliveryHandler interface {
	AmqpMessageHandler(message []byte) error
}

type Consumer struct {
	*client
	queue          *amqp.Queue
	ctag           string
	deliveryCount  int64
	wg             sync.WaitGroup
	messageHandler DeliveryHandler
}

func NewConsumer(handler DeliveryHandler) *Consumer {
	return &Consumer{
		client:         &client{},
		messageHandler: handler,
	}
}

func (c *Consumer) Create(url, exchange, kind, queue_name, binding_key string) error {
	err := c.create(url, exchange, kind)
	if err != nil {
		return err
	}
	channel := c.channel
	//声明队列
	queue, err := channel.QueueDeclare(
		queue_name, // name of the queue
		true,       // durable
		false,      // delete when unused
		false,      // exclusive
		false,      // noWait
		nil,        // arguments
	)
	if err != nil {
		log.Error("AMQP queue declare error: %s", err.Error())
		return err
	}

	//将队列绑定到交换器上
	if err = channel.QueueBind(
		queue.Name,  // name of the queue
		binding_key, // bindingKey
		exchange,    // sourceExchange
		false,
		nil, // arguments
	); err != nil {
		log.Error("queue Bind err: %s", err.Error())
		return err
	}
	log.Info("AMQP declared queue (%q %d messages, %d consumers), binding to exchange (key %q)", queue.Name, queue.Messages, queue.Consumers, binding_key)
	c.queue = &queue
	return nil
}

func (c *Consumer) Destroy() error {
	// will close() the deliveries channel
	if err := c.channel.Cancel(c.ctag, true); err != nil {
		log.Error("AMQP consumer cancel failed: %s", err.Error())
		return err
	}
	if err := c.destroy(); err != nil {
		return err
	}
	log.Info("AMQP shutdown OK,and receive deliveries:%d", c.deliveryCount)
	c.wg.Wait()
	return nil
}

func (c *Consumer) StartSubscribe(ctx context.Context, ctag string, autoAck bool) error {
	//log.Info("queue bound to exchange, starting consume (consumer tag %s)", ctag)
	//使用推模式消费消息,通过持续订阅的方式来消费消息
	channel := c.channel
	queue := c.queue
	//消费者流控  防止数据库爆库
	//消息的消费需要配合Qos
	if err := channel.Qos(
		//每次队列只消费一个消息 这个消息处理不完服务器不会发送第二个消息过来
		//当前消费者一次能接受的最大消息数量
		1,
		//服务器传递的最大容量
		0,
		//如果为true 对channel可用 false则只对当前队列可用
		false,
	); err != nil {
		log.Error("AMQP channel qos err:%s", err.Error())
		return err
	}
	//开始接受消息
	deliveries, err := channel.Consume(
		queue.Name, // name
		ctag,       // consumerTag,
		autoAck,    // autoAck
		false,      // exclusive
		false,      // noLocal
		false,      // noWait
		nil,        // arguments
	)
	if err != nil {
		log.Error("AMQP queue consume: %s", err.Error())
		return err
	}

	go func() {
		c.wg.Add(1)
		c.handleDeliveries(deliveries, autoAck)
		c.wg.Done()
	}()
	return nil
}

func (c *Consumer) handleDeliveries(deliveries <-chan amqp.Delivery, autoAck bool) {
	for d := range deliveries {
		c.deliveryCount++
		log.Info("AMQP got %d B delivery: [%v] %s", len(d.Body), d.DeliveryTag, string(d.Body))
		if err := c.handleMessage(d.Body); err != nil {
			log.Error("AMQP handle message:%s err:%s", err.Error())
		}
		if !autoAck {
			//如果为true表示确认所有未确认的消息,如果为false表示确认当前消息
			//执行完业务逻辑成功之后我们再手动ack告诉服务器你可以删除这个消息啦！ 这样就保障了数据的绝对的安全不丢失！
			if err := d.Ack(false); err != nil {
				log.Error("AMQP ack err:%s", err.Error())
			}
		}
	}
	log.Info("AMQP handle: deliveries channel closed")
}

func (c *Consumer) handleMessage(message []byte) error {
	if c.messageHandler != nil {
		return c.messageHandler.AmqpMessageHandler(message)
	}
	return nil
}
